package event

import (
	"github.com/618lf/swakx-go/pkg/pool"
	"reflect"
	"sync"
)

// Callback is called when the subscribed event happens.
type Callback func(*Event)

// Bus implements the observer pattern to allow event dispatching and watching.
type Bus interface {
	Watch(event string, callback Callback)
	UnWatch(event string, callback Callback)
	Dispatch(event *Event)
	DispatchAndWait(event *Event)
}

// bus is implements Bus
type bus struct {
	Bus
	callbacks sync.Map
	wpool     pool.WorkerPool
}

// Watch subscribe to a certain event with a callback.
func (b *bus) Watch(event string, callback Callback) {
	var callbacks []Callback
	if actual, ok := b.callbacks.Load(event); ok {
		callbacks = append(callbacks, actual.([]Callback)...)
	}
	callbacks = append(callbacks, callback)
	b.callbacks.Store(event, callbacks)
}

// UnWatch remove the given callback from the callback chain of an event.
func (b *bus) UnWatch(event string, callback Callback) {
	var filtered []Callback

	// In go, functions are not comparable, so we use reflect.ValueOf(callback).Pointer() to reflect their address for comparison.
	target := reflect.ValueOf(callback).Pointer()
	if actual, ok := b.callbacks.Load(event); ok {
		for _, h := range actual.([]Callback) {
			if reflect.ValueOf(h).Pointer() != target {
				filtered = append(filtered, h)
			}
		}
		b.callbacks.Store(event, filtered)
	}
}

// Dispatch dispatches an event by invoking each callback asynchronously.
func (b *bus) Dispatch(event *Event) {
	if actual, ok := b.callbacks.Load(event.Name); ok {
		for _, h := range actual.([]Callback) {
			f := h // assign the value to a new variable for the closure
			b.wpool.Execute(func() {
				f(event)
			})
		}
	}
}

// DispatchAndWait dispatches an event by invoking callbacks concurrently and waits for them to finish.
func (b *bus) DispatchAndWait(event *Event) {
	if actual, ok := b.callbacks.Load(event.Name); ok {
		for _, h := range actual.([]Callback) {
			f := h // assign the value to a new variable for the closure
			future := b.wpool.Submit(func() {
				f(event)
			})
			future.Get()
		}
	}
}

// Config config for  Bus
type Config struct {
	PoolSize int
	Pool     pool.WorkerPool
}

func (c *Config) WithPoolSize(poolSize int) *Config {
	c.PoolSize = poolSize
	return c
}

func (c *Config) WithPool(pool pool.WorkerPool) *Config {
	c.Pool = pool
	return c
}
func (c *Config) Build() Bus {
	wp := c.Pool
	if wp == nil {
		wp = pool.NewWorkerPool(c.PoolSize)
	}
	return &bus{
		wpool: wp,
	}
}

func NewEventBusBuilder() *Config {
	return &Config{}
}

func NewEventBus(config *Config) Bus {
	if config == nil || config.PoolSize <= 0 || config.Pool == nil {
		panic("config error!")
	}

	wp := config.Pool
	if wp == nil {
		wp = pool.NewWorkerPool(config.PoolSize)
	}
	return &bus{
		wpool: wp,
	}
}
